Setting up Yeedu with Prefect

In this guide we walk you through integrating Yeedu with Prefect, step by step, to create and manage data workflows.

Installation

To use the Yeedu Operator in your Prefect environment, install it using the following command:

pip3 install prefect-yeedu-operator

Overview

The YeeduOperator enables you to efficiently define and orchestrate Yeedu jobs and notebooks. We built it to simplify the process by:

• Submitting Jobs and Notebooks: Scheduling jobs and notebooks in Yeedu from a Prefect task.
• Monitoring: The operator keeps you informed about the status of your submitted Yeedu jobs and notebooks, providing you with real-time updates.
• Handling Completion: After execution, the operator accurately captures and manages the status updates of Yeedu jobs and notebooks.
• Managing Logs: All relevant logs associated with Yeedu jobs and notebooks are accessible within Prefect.

1. Prerequisites

Before using PrefectYeeduOperator, make sure you've met the following requirements:

• Access to Yeedu
• Creating blocks via the Prefect UI: To store Yeedu credentials
• Authenticate to Prefect Cloud via CLI: Ensure you're authenticated to Prefect Cloud from the CLI before submitting flows
• Install YeeduOperator: To use the Yeedu Operator in your Prefect environment, install prefect-yeedu-operator

1.1 Create Prefect blocks

Blocks can be created in two ways

• Prefect UI
• Python code

Using Prefect UI

1. Accessing the Blocks Section

   • Navigate to the Prefect UI and click on Configurations in the left panel.
   • From dropdown menu, select the Blocks section .

2. Creating a JSON block

   • Click on the plus symbol + to create new connection.

   • Search for JSON and click on Create. Fill in the required details

     • Block Name: Provide a unique name. Example: yeedu_connection_details
     • Value: Provide below JSON

     {
         "username": "YSU0000",
         "YEEDU_VERIFY_SSL": "false",
         "YEEDU_SSL_CERT_FILE": "Provide the path to the certificate file if YEEDU_VERIFY_SSL is true" 
     }

3. Creating a Secret block

   • Click on the plus symbol + to create new secret.
   • Search for Secret and click on Create. Fill in the required details
     • Block Name: Provide a unique name. Example: block_login_password
     • Value: Provide Yeedu login password or token

   Note: If the Yeedu authentication type is

   • LDAP or Azure AD: create a block for storing password.
   • Azure AD SSO: create a block for storing token.

Using python

1. Create and Save a JSON Block

   Add the following lines of code to a python file.

   • value - Modify the JSON value details as per your requirement

   • name - block_name

     from prefect.blocks.system import JSON
     json_block = JSON(value={
         "username":"ysu0000",
         "YEEDU_VERIFY_SSL":"false",
         "YEEDU_SSL_CERT_FILE":"Provide the path to the certificate file if      YEEDU_VERIFY_SSL is true"})
     json_block.save(name="connection-block") 
     

2. Create and Save a Secret Block

   Add the following lines of code to a python file.

   • value - Replace <Password> with the required value

   • name - block_name

     from prefect.blocks.system import Secret
     secret_block = Secret(value="<Password>")
     secret_block.save(name="secret-block") 

3. Update saved block value

   To overwrite the existing block value, pass overwrite=True

       json_block.save(name="connection-block",overwrite=True)

4. Run python file

   python3 create_block.py

For detailed guidance on managing blocks, please visit Blocks Documentation

Note- Before creating blocks using Python code, ensure that your environment is authenticated to Prefect Cloud

1.2 Authenticate to Prefect Cloud via CLI

See the API Keys documentation to generate an API key from the Prefect UI.

prefect cloud login -k <API_KEY>

2. Flow Creation

We designed the YeeduOperator to be instantiated inside a Prefect task, which the flow then calls. Use the code below to schedule a job or notebook.

• job_url: The URL of the Yeedu notebook or job. It must include the Yeedu REST API port, for example https://hostname:8080/tenant/<tenant_id>/workspace/<workspace_id>/job/<job_id>.
• connection_block_name - connection details
• login_password_block_name - password

from prefect import flow, task
from operators.yeedu import YeeduOperator

CONNECTION_BLOCK = '<NAME_OF_CONNECTION_BLOCK_VARIABLE>'
PASSWORD_BLOCK = '<NAME_OF_PASSWORD_BLOCK_VARIABLE>'
JOB_URL = '<YEEDU_JOB_URL>'


@task
def run_yeedu_job(job_url):
    # Initialize and execute the Yeedu operator
    YeeduOperator(
        job_url=job_url,
        connection_block_name=CONNECTION_BLOCK,
        login_password_block_name=PASSWORD_BLOCK,
        # Uncomment the following line to use a token block for SSO sign-in
        # token_block_name='<NAME_OF_TOKEN_BLOCK>',
    ).execute()


@flow(retries=0, retry_delay_seconds=5, log_prints=True)
def flow_name():
    run_yeedu_job(JOB_URL)

Operator Parameters

Required

Parameter	Type	Description
job_url	str	URL of the Yeedu job, notebook, or health check, including the REST API port. Encodes the tenant ID, workspace ID, run type and conf ID.
connection_block_name	str	Name of the Prefect JSON block holding username, YEEDU_VERIFY_SSL and YEEDU_SSL_CERT_FILE.
login_password_block_name	str	Name of the Prefect Secret block holding the login password.

Optional

Parameter	Type	Description
token_block_name	str	Name of a Prefect Secret block holding a Yeedu SSO session token. When provided, the token is used for authentication instead of the password.
arguments	str	Arguments passed to the job or notebook run.
conf	List[str]	Spark/runtime configuration overrides in key=value format. Must be a list. If the same key appears twice, the last value wins.
cluster_ids	List[int]	Fallback cluster IDs to retry on when the run fails due to resource exhaustion (OOM / exit 137). The job always runs on its configured cluster first, then tries these in order. Duplicates are removed automatically.
email_recipients	List[str]	Email addresses to notify when the task succeeds or fails. Requires the email Secret blocks to be configured.

Example with optional parameters

@task
def run_yeedu_job(job_url):
    YeeduOperator(
        job_url=job_url,
        connection_block_name=CONNECTION_BLOCK,
        login_password_block_name=PASSWORD_BLOCK,
        token_block_name='<NAME_OF_TOKEN_BLOCK>',
        arguments='--date 2024-01-01',
        conf=[
            'spark.driver.memory=4g',
            'spark.executor.memory=8g',
        ],
        cluster_ids=[10, 20, 30],
        email_recipients=['user@example.com'],
    ).execute()

note

Prefect doesn't have XCom. The Airflow operator's loop_input parameter has no Prefect equivalent, so pass dynamic per-run values as Prefect task parameters and return them from the task instead.

3. Execute the Flow - Creating Deployment

3.1 Deployment with Local storage

This deployment mode uses local machine as Prefect Worker and Prefect Flow Code stored Locally

• Add the following lines to your flow code

  if __name__ == "__main__":
      
      flow_name.serve(name="<NAME_OF_DEPLOYMENT>")

• Run the python file

  python3 file_name.py 

3.2 Deployment with Remote Storage

This deployment mode uses Prefect Work pools with Prefect Flow Code stored in Remote Github Repository

Use the following code snippet.

• url - The URL of the GitHub repository.
• branch - provide branch name
• access_token - provide block name .
• name - The name of the deployment
• work_pool_name - The name of the work pool.
• entrypoint - Specify the path to the flow code file stored in GitHub along with the name of the flow

from prefect import flow
from prefect.runner.storage import GitRepository
from prefect.blocks.system import Secret

flow.from_source(
    source=GitRepository(
        url="<YOUR_GITHUB_REPO_URL>",
        branch="<YOUR_BRANCH_NAME>",
        credentials={
            "access_token": Secret.load("<NAME_OF_THE_GITHUB_TOKEN_BLOCK_VARIABLE>")
        }
    ),
    entrypoint="<YOUR_ENTRYPOINT_FILE>:<YOUR_FLOW_FUNCTION>",
).deploy(
    name="<NAME_OF_THE_DEPLOYMENT>",
    work_pool_name="<NAME_OF_WORKER>"
)